﻿/*
   Copyright 2012 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reflection;
using System.ServiceModel;
using System.ServiceProcess;
using System.Text;
using System.Runtime.Remoting.Lifetime;
using System.Threading;
using System.Threading.Tasks;
using System.Transactions;
using C5;
using Flower.Actors;
using Flower.Client;
using Flower.Logging;
using Flower.Services;
using Flower.Services.Data;
using Assembly = System.Reflection.Assembly;
using Flower.Directory.Util;
using IsolationLevel = System.Transactions.IsolationLevel;
using Process = Flower.Services.Data.Process;
using System.Security.Principal;
using System.Security;
using Flower.Services.Clients;

namespace Flower.Processor
{
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
    public partial class ProcessorService : ServiceBase, IProcessingManager, IProcessor
    {
        [Flags]
        public enum Options
        {
            None = 0,
            DoNotStartServiceHost = 1,
            DoNotCastAppDomains = 2
        }

        private class Suspention
        {
            public string Pid;
            public DateTime Until;
            public string LinkId;
        }

        private class SuspentionComparer : IComparer<Suspention>
        {
            public static readonly SuspentionComparer Instance = new SuspentionComparer();

            public int Compare(Suspention x, Suspention y)
            {
                return DateTime.Compare(x.Until, y.Until);
            }
        }

        private class ProcessorHostStatistics
        {
            public long ProcessesCount;
            public readonly System.Collections.Generic.HashSet<string> WaitingProcesses = new System.Collections.Generic.HashSet<string>();
            public DateTime? LastTimeProcessFinished;
        }

        private readonly string _pendingFolderPath;
        private readonly string _runningFolderPath;
        private readonly string _waitingFolderPath;
        private readonly string _finishedFolderPath;
        private readonly string _brokenFolderPath;

        private static readonly TimeSpan ProcessorHostsIdlenessCheckInterval = TimeSpan.FromMinutes(5);
        private static readonly TimeSpan ProcessorHostMaxIdleTime = TimeSpan.FromMinutes(5);
        private static readonly TimeSpan SuspentionCheckInterval = TimeSpan.FromSeconds(1);
        private static readonly TimeSpan RegistrationCheckInterval = TimeSpan.FromSeconds(30);
        private static readonly TimeSpan MaxActorWaitTime = TimeSpan.FromSeconds(30);

        private const int PROCESSES_CHECK_BATCH_SIZE = 100;
        
        private static readonly Log Log = LogManager.CreateLog(typeof(ProcessorService).Name);

        private readonly IFlowerClient _flowerClient = ApplicationContext.GetDirectoryClient();

        private IActor<Command> _actor; 

        private readonly Dictionary<string, IProcessorHost> _processorHosts =
            new Dictionary<string, IProcessorHost>();
        private readonly Dictionary<string, string> _pidToAppDomainMap = 
            new Dictionary<string, string>(); 
        private readonly ClientSponsor _sponsor = new ClientSponsor();
        private readonly Dictionary<string, ProcessorHostStatistics> _processorHostStatistics =
            new Dictionary<string, ProcessorHostStatistics>();

        private Thread _suspentionsCheckerThread;
        private readonly ArrayHeap<Suspention> _suspentions =
            new ArrayHeap<Suspention>(SuspentionComparer.Instance);
        private readonly ReaderWriterLockSlim _suspentionsLock =
            new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
        private readonly ManualResetEventSlim _suspentionStopEvent = new ManualResetEventSlim();

        private readonly ProcessorSettings _settings = ApplicationContext.GetProcessorSettings();
        
        private Options _options;
        private ServiceHost _serviceHost;
        private Timer _idlenessCheckTimer;
        private Timer _pendingProcessesCheckingTimer;
        private bool _pendingDebt = false;
        private DateTime _lastPendingProcessesCheck = DateTime.Now;

        private volatile Timer _registrationCheckTimer;
        private volatile bool _isInitialized;
        private volatile bool _isStopped;
        private volatile bool _isSuspended;
        private readonly object _registrationCheckTimerLock = new object();
        private readonly object _initializationLock = new object();

        public ProcessorService()
        {
            InitializeComponent();

            _pendingFolderPath = string.Concat("/Processors/", _settings.ProcessorName, "/Pending");
            _runningFolderPath = string.Concat("/Processors/", _settings.ProcessorName, "/Running");
            _waitingFolderPath = string.Concat("/Processors/", _settings.ProcessorName, "/Waiting");
            _finishedFolderPath = string.Concat("/Processors/", _settings.ProcessorName, "/Finished");
            _brokenFolderPath = string.Concat("/Processors/", _settings.ProcessorName, "/Broken");
        }

        #region Start/stop

        protected override void OnStart(string[] args)
        {
            DoStart(Options.None);
        }

        internal void DoStart(Options options)
        {
            Log.Info("Starting service.");

            _options = options;

            try
            {
                if (!EnsureInitialized())
                {
                    _registrationCheckTimer = new Timer
                    (
                        _ => 
                        {
                            try
                            {
                                if (EnsureInitialized())
                                {
                                    Task.Factory.StartNew(() =>
                                    {
                                        try
                                        {
                                            lock (_registrationCheckTimerLock)
                                            {
                                                if (_registrationCheckTimer != null)
                                                {
                                                    _registrationCheckTimer.Dispose();
                                                    _registrationCheckTimer = null;
                                                }
                                            }
                                        }
                                        catch (Exception ex)
                                        {
                                            Log.Error("Failed to dispose the registration check timer.", ex);
                                        }
                                    });
                                }
                            }
                            catch (Exception ex)
                            {
                                Log.Error("Failed to check processor registration.", ex);
                            }
                        }, 
                        null, 
                        RegistrationCheckInterval,
                        RegistrationCheckInterval
                    );
                }

                if (!options.HasFlag(Options.DoNotStartServiceHost))
                {
                    _serviceHost = new ServiceHost(this);
                    _serviceHost.Open();
                }
                else
                {
                    _serviceHost = null;
                }
                
                Log.Info("Service started.");
            }
            catch (Exception ex)
            {
                Log.Fatal("Failed to start the service.", ex);
                throw;
            }
        }

        private bool EnsureInitialized()
        {
            if (!_isInitialized)
            {
                lock (_initializationLock)
                {
                    if (!_isInitialized)
                    {
                        if (_isStopped)
                        {
                            return false;
                        }

                        try
                        {
                            if (!IsDirectoryStructureValid())
                            {
                                Log.Warning(string.Format("Failed to initialize processor {0}. The processor is not registered in the directory.  Will be retried at {1}.", _settings.ProcessorName, DateTime.Now + RegistrationCheckInterval));
                                return false;
                            }

                            _actor = new Actor<ProcessorService, Command>
                            (
                                this,
                                new PriorityQueue<Command>(CommandComparer.Instance)
                            );
                            _actor.Error += (sender, ev) =>
                                Log.Error
                                (
                                    string.Format
                                    (
                                        "Error occurred while processing the command {0}.",
                                        ev.Message.GetType().Name
                                    ),
                                    ev.Exception
                                );

                            _pendingProcessesCheckingTimer =
                                new Timer
                                (
                                    state => CheckPendingProcesses(false),
                                    null,
                                    _settings.PendingProcessesCheckingInterval,
                                    _settings.PendingProcessesCheckingInterval
                                );
                            _idlenessCheckTimer =
                                new Timer
                                (
                                    state => _actor.Send(CheckProcessorHostsIdlenessCommand.Instance),
                                    null,
                                    ProcessorHostsIdlenessCheckInterval,
                                    ProcessorHostsIdlenessCheckInterval
                                );

                            ResumeRunningProcesses();
                            CheckWaitingProcesses();
                            CheckSuspendedProcesses();

                            _suspentionsCheckerThread = new Thread(ProcessSuspentions)
                            {
                                IsBackground = true
                            };
                            _suspentionsCheckerThread.Start();

                            _isInitialized = true;
                        }
                        catch (WcfClientException ex)
                        {
                            Log.Warning(string.Format("Failed to initialize processor {0} due to directory communication error '{1}'. Will be retried at {2}.", _settings.ProcessorName, ex.InnerException.Message, DateTime.Now + RegistrationCheckInterval));
                            return false;
                        }
                        catch (Exception ex)
                        {
                            Log.Error(string.Format("Failed to initialize processor {0}. Will be retried at {1}.", _settings.ProcessorName, DateTime.Now + RegistrationCheckInterval), ex);
                            return false;
                        }
                    }
                }
            }

            return _isInitialized;
        }

        private bool IsDirectoryStructureValid()
        {
            const string validationScript = 
                "$out.isDirStructureValid = " +
                "  $dir.exists($in.pendingFolderPath) &&" +
                "  $dir.exists($in.runningFolderPath) &&" +
                "  $dir.exists($in.waitingFolderPath) &&" +
                "  $dir.exists($in.finishedFolderPath) &&" +
                "  $dir.exists($in.brokenFolderPath);";

            var result =
                TransactionHelper.Scope
                (
                    IsolationLevel.ReadCommitted,
                    TimeSpan.FromSeconds(30),
                    () =>
                        _flowerClient
                            .CreateScript()
                            .SetText(validationScript)
                            .AddParameter("pendingFolderPath", _pendingFolderPath)
                            .AddParameter("runningFolderPath", _runningFolderPath)
                            .AddParameter("waitingFolderPath", _waitingFolderPath)
                            .AddParameter("finishedFolderPath", _finishedFolderPath)
                            .AddParameter("brokenFolderPath", _brokenFolderPath)
                            .Execute()
                );

            return result.Get<bool>("isDirStructureValid");
        }

        protected override void OnStop()
        {
            DoStop();
        }

        internal void DoStop()
        {
            Log.Info("Stopping service.");

            if (_serviceHost != null)
            {
                _serviceHost.Close();
                Log.Trace("Service host closed.");
            }

            lock (_registrationCheckTimerLock)
            {
                if (_registrationCheckTimer != null)
                {
                    _registrationCheckTimer.Dispose();
                    Log.Trace("Registration checker stopped.");
                }
            }

            lock (_initializationLock)
            {
                if (_isInitialized)
                {
                    _idlenessCheckTimer.Dispose();

                    _suspentionStopEvent.Set();
                    _suspentionsCheckerThread.Join();

                    Log.Trace("Suspentions checker stopped.");

                    if (_actor != null)
                    {
                        _actor.Stop();
                        Log.Trace("Actor stopped.");
                    }

                    StopAllProcessors();
                    Log.Trace("Processors stopped.");

                    if (_actor != null)
                    {
                        _actor.Dispose();
                    }
                }

                _isStopped = true;
            }

            Log.Info("Service is stopped.");
        }

        #endregion

        #region Suspention checking

        private void ProcessSuspentions(object args)
        {
            Log.Trace("Entering the suspentions checking thread.");

            TimeSpan waitTime;

            do
            {
                if (_isSuspended)
                {
                    waitTime = SuspentionCheckInterval;
                    continue;
                }

                _suspentionsLock.EnterUpgradeableReadLock();
                try
                {
                    if (_suspentions.Count > 0)
                    {
                        Suspention nextToResume = _suspentions.Peek();
                        var now = DateTime.Now;

                        if (now >= nextToResume.Until)
                        {
                            _suspentionsLock.EnterWriteLock();

                            //TODO: Optimize - put all suspentions to resume into a batch.

                            //We need to read the min again because it may already be another
                            //object, but the priority policy guarantees that the if-condition
                            //holds for it.
                            nextToResume = _suspentions.Peek();
                            
                            try
                            {
                                Log.Trace(() => string.Format("Resuming suspended process {0} at {1}.", nextToResume.Pid, now));

                                TransactionHelper.Scope
                                (
                                    IsolationLevel.ReadCommitted,
                                    TimeSpan.FromSeconds(30),
                                    () => 
                                    {
                                        _flowerClient
                                            .CreateScript()
                                            .SetText("@ChangeProcessStatus")
                                            .AddParameter("pid", nextToResume.Pid)
                                            .AddParameter("status", "Pending")
                                            .Execute();

                                        _flowerClient.Directory.Remove(nextToResume.LinkId);
                                    }
                                );
                            }
                            catch (Exception ex)
                            {
                                Log.Error(string.Format("Failed to resume the process {0}.", nextToResume.Pid), ex);

                                try
                                {
                                    TransactionHelper.Scope
                                    (
                                        IsolationLevel.ReadCommitted,
                                        TimeSpan.FromSeconds(30),
                                        () =>
                                            _flowerClient
                                                .CreateScript()
                                                .SetText("@ChangeProcessStatus")
                                                .AddParameter("pid", nextToResume.Pid)
                                                .AddParameter("status", "Broken")
                                                .Execute()
                                    );
                                }
                                catch(Exception e)
                                {
                                    Log.Error(string.Format("Failed to mark the process {0} broken.", nextToResume.Pid), e);
                                }
                            }
                            finally
                            {
                                waitTime = TimeSpan.Zero;
                                _suspentions.Pop();
                                _suspentionsLock.ExitWriteLock();
                            }
                        }
                        else
                        {
                            waitTime = SuspentionCheckInterval;
                        }
                    }
                    else
                    {
                        waitTime = SuspentionCheckInterval;
                    }
                }
                catch (Exception ex)
                {
                    Log.Error("Failed to resume suspended proceses.", ex);
                    waitTime = SuspentionCheckInterval;
                }
                finally
                {
                    _suspentionsLock.ExitUpgradeableReadLock();
                }
            }
            while (!_suspentionStopEvent.Wait(waitTime));
        }

        #endregion

        #region Commands

        private class CommandComparer : IComparer<Command>
        {
            public static readonly CommandComparer Instance = new CommandComparer();

            public int Compare(Command x, Command y)
            {
                int eq = x.Priority - y.Priority;

                if (eq == 0)
                {
                    eq = -DateTime.Compare(x.Timestamp, y.Timestamp);
                }

                return eq;
            }
        }

        private abstract class Command : IMessage<ProcessorService>
        {
            public DateTime Timestamp;

            public void Init()
            {
                Timestamp = DateTime.Now;
            }

            public abstract int Priority { get; }
            public abstract void Process(ProcessorService processor);
        }

        private class MarkWaitingCommand : Command
        {
            public string Pid;

            public override int Priority
            {
                get { return 1; }
            }

            public override void Process(ProcessorService processor)
            {
                processor.Process(this);
            }
        }

        private class MarkRunningCommand : Command
        {
            public string Pid;

            public override int Priority
            {
                get { return 1; }
            }

            public override void Process(ProcessorService processor)
            {
                processor.Process(this);
            }
        }

        private class CheckPendingProcessesCommand : Command
        {
            public bool ResumeActors;

            public override int Priority
            {
                get { return 2; }
            }

            public override void Process(ProcessorService processor)
            {
                processor.Process(this);
            }
        }
        private class MarkBrokenCommand : Command
        {
            public string Pid;

            public override int Priority
            {
                get { return 3; }
            }

            public override void Process(ProcessorService processor)
            {
                processor.Process(this);
            }
        }

        private class MarkFinishedCommand : Command
        {
            public string Pid;

            public override int Priority
            {
                get { return 4; }
            }

            public override void Process(ProcessorService processor)
            {
                processor.Process(this);
            }
        }

        private class CheckProcessorHostsIdlenessCommand : Command
        {
            public static readonly CheckProcessorHostsIdlenessCommand Instance =
                new CheckProcessorHostsIdlenessCommand();

            public override int Priority
            {
                get { return 5; }
            }

            public override void Process(ProcessorService processor)
            {
                processor.Process(this);
            }
        }

        private readonly ObjectPool<MarkWaitingCommand> _markWaitingCommandPool =
            new ObjectPool<MarkWaitingCommand>(cmd => cmd.Init(), cmd => cmd.Pid = null);
        private readonly ObjectPool<MarkRunningCommand> _markRunningCommandPool =
            new ObjectPool<MarkRunningCommand>(cmd => cmd.Init(), cmd => cmd.Pid = null);
        private readonly ObjectPool<CheckPendingProcessesCommand> _checkPendingProcessesCommandPool =
            new ObjectPool<CheckPendingProcessesCommand>(cmd => cmd.Init(), null);
        private readonly ObjectPool<MarkBrokenCommand> _markBrokenCommandPool =
            new ObjectPool<MarkBrokenCommand>(cmd => cmd.Init(), cmd => cmd.Pid = null);
        private readonly ObjectPool<MarkFinishedCommand> _markFinishedCommandPool =
            new ObjectPool<MarkFinishedCommand>(cmd => cmd.Init(), cmd => cmd.Pid = null);

        #endregion

        #region Command handlers

        private void Process(MarkFinishedCommand command)
        {
            Log.Trace(() => string.Format("Process {0} is marked finished.", command.Pid));

            try
            {
                string appDomain;

                if (_pidToAppDomainMap.TryGetValue(command.Pid, out appDomain))
                {
                    TransactionHelper.Scope
                    (
                        IsolationLevel.ReadCommitted,
                        TimeSpan.FromSeconds(30),
                        () =>
                            _flowerClient
                                .CreateScript()
                                .SetText("@ChangeProcessStatus")
                                .AddParameter("pid", command.Pid)
                                .AddParameter("status", "Finished")
                                .Execute()
                    );

                    RemoveProcessFromStatistics(appDomain, command.Pid, false);
                    _pidToAppDomainMap.Remove(command.Pid);
                }
            }
            finally
            {
                _markFinishedCommandPool.Return(command);
            }
        }

        private void Process(MarkBrokenCommand command)
        {
            Log.Trace(() => string.Format("Process {0} is marked broken.", command.Pid));

            try
            {
                string appDomain;

                if (_pidToAppDomainMap.TryGetValue(command.Pid, out appDomain))
                {
                    TransactionHelper.Scope
                    (
                        IsolationLevel.ReadCommitted,
                        TimeSpan.FromSeconds(30),
                        () =>
                            _flowerClient
                                .CreateScript()
                                .SetText("@ChangeProcessStatus")
                                .AddParameter("pid", command.Pid)
                                .AddParameter("status", "Broken")
                                .Execute()
                    );

                    RemoveProcessFromStatistics(appDomain, command.Pid, false);
                    _pidToAppDomainMap.Remove(command.Pid);
                }
            }
            finally
            {
                _markBrokenCommandPool.Return(command);
            }
        }

        private void Process(MarkRunningCommand command)
        {
            Log.Trace(() => string.Format("Process {0} is marked running.", command.Pid));

            try
            {
                AddProcessToStatistics(_pidToAppDomainMap[command.Pid], command.Pid, false);
            }
            finally
            {
                _markRunningCommandPool.Return(command);
            }
        }

        private void Process(MarkWaitingCommand command)
        {
            Log.Trace(() => string.Format("Process {0} is marked waiting.", command.Pid));

            try
            {
                RemoveProcessFromStatistics(_pidToAppDomainMap[command.Pid], command.Pid, true);
            }
            finally
            {
                _markWaitingCommandPool.Return(command);
            }
        }

        private bool CanTakeMoreProcesses(long processesCount)
        {
            return 
                _settings.MaxProcesses == 0 ||
                processesCount < _settings.MaxProcesses;
        }

        private long GetProcessesCount()
        {
            return _processorHostStatistics.Sum(stat => stat.Value.ProcessesCount);
        }

        private void Process(CheckPendingProcessesCommand command)
        {
            Log.Trace("Checking pending processes.");

            try
            {
                if (command.Timestamp <= _lastPendingProcessesCheck)
                {
                    Log.Trace("Skipping exceeding pending processors check.");
                    return;
                }

                _lastPendingProcessesCheck = command.Timestamp;

                long totalProcessesTaken = 0;
                long currentProcessesCount = GetProcessesCount();
                bool canTakeMoreProcesses = CanTakeMoreProcesses(currentProcessesCount);

                _pendingDebt = false;

                Log.Trace(() => string.Format("Current processes count {0}.", currentProcessesCount));

                if (!canTakeMoreProcesses)
                {
                    Log.Trace("Can't take more processes.");

                    _pendingDebt = true;

                    if (command.ResumeActors)
                    {
                        foreach (IProcessorHost host in _processorHosts.Values)
                        {
                            host.ResumeActors();
                        }
                    }
                    return;
                }

                var processesTaken = new List<DirectoryEntry>();
                System.Collections.Generic.IList<DirectoryEntry> pendingProcesses = null;
                var transactionOptions = new TransactionOptions
                {
                    Timeout = TimeSpan.FromSeconds(30),
                    IsolationLevel = IsolationLevel.ReadCommitted
                };

                do
                {
                    using (var txn = new TransactionScope(TransactionScopeOption.Required, transactionOptions))
                    {
                        pendingProcesses =
                            _flowerClient.Directory.GetChildren
                            (
                                _pendingFolderPath,
                                DirectoryEntryTypes.Process,
                                SortProperty.Default,
                                SortOrder.Asc,
                                0,
                                PROCESSES_CHECK_BATCH_SIZE
                            );

                        processesTaken.Clear();

                        foreach (Process pendingProcess in pendingProcesses)
                        {
                            processesTaken.Add(pendingProcess);
                            ++totalProcessesTaken;

                            if (!(canTakeMoreProcesses = CanTakeMoreProcesses(currentProcessesCount + totalProcessesTaken)))
                            {
                                Log.Trace("Can't take more processes.");
                                _pendingDebt = true;
                                break;
                            }
                        }

                        if (processesTaken.Count > 0)
                        {
                            _flowerClient
                                .CreateScript()
                                .SetText("@ChangeProcessStatuses")
                                .AddParameter("pids", processesTaken.Select(p => p.Id).ToArray())
                                .AddParameter("status", "Running")
                                .Execute();
                        }

                        txn.Complete();
                    }

                    foreach (Process pendingProcess in processesTaken)
                    {
                        try
                        {
                            RegisterProcess(pendingProcess, false);
                        }
                        catch (Exception ex)
                        {
                            Log.Error(string.Format("Failed to load the process {0}.", pendingProcess.Id), ex);

                            TransactionHelper.Scope
                            (
                                IsolationLevel.ReadCommitted,
                                TimeSpan.FromSeconds(30),
                                () =>
                                    _flowerClient
                                        .CreateScript()
                                        .SetText("@ChangeProcessStatus")
                                        .AddParameter("pid", pendingProcess.Id)
                                        .AddParameter("status", "Broken")
                                        .Execute()
                            );
                        }
                    }
                }
                while (pendingProcesses.Count > 0 && canTakeMoreProcesses);

                Log.Trace(() => string.Format("Took {0} processes.", totalProcessesTaken));

                if (command.ResumeActors)
                {
                    foreach (IProcessorHost host in _processorHosts.Values)
                    {
                        host.ResumeActors();
                    }
                }
            }
            finally
            {
                _checkPendingProcessesCommandPool.Return(command);
            }
        }

        private void Process(CheckProcessorHostsIdlenessCommand command)
        {
            Log.Trace("Checking processor hosts idleness.");

            foreach (var stat in _processorHostStatistics.ToList())
            {
                if
                (
                    stat.Value.LastTimeProcessFinished.HasValue &&
                    stat.Value.ProcessesCount == 0 &&
                    (DateTime.Now - stat.Value.LastTimeProcessFinished.Value) > ProcessorHostMaxIdleTime
                )
                {
                    string appDomainName = stat.Key;

                    Log.Trace(() => string.Format("Processor host {0} has been expired.", appDomainName));

                    StopProcessorHost(appDomainName);
                }
            }
        }

        private static string AppDomainNameByAssemblyName(AssemblyName assemblyName)
        {
            return string.Concat("ProcessorHost: ", assemblyName.Name, "_", assemblyName.Version);
        }

        private void AddProcessToStatistics(string appDomainName, string pid, bool isWaiting)
        {
            ProcessorHostStatistics stat;

            if (!_processorHostStatistics.TryGetValue(appDomainName, out stat))
            {
                stat = new ProcessorHostStatistics();
                _processorHostStatistics.Add(appDomainName, stat);
            }

            if (isWaiting)
            {
                stat.WaitingProcesses.Add(pid);
            }
            else
            {
                ++stat.ProcessesCount;
                stat.WaitingProcesses.Remove(pid);
            }
        }

        private void RemoveProcessFromStatistics(string appDomainName, string pid, bool isWaiting)
        {
            ProcessorHostStatistics stat;

            if (_processorHostStatistics.TryGetValue(appDomainName, out stat))
            {
                --stat.ProcessesCount;

                if (isWaiting)
                {
                    stat.WaitingProcesses.Add(pid);
                }
                else
                {
                    stat.LastTimeProcessFinished = DateTime.Now;
                }

                if (_pendingDebt)
                {
                    CheckPendingProcesses(false);
                }
            }
        }

        private IProcessorHost GetOrCreateProcessorHost(string appDomainName)
        {
            IProcessorHost processorHost;

            if (!_processorHosts.TryGetValue(appDomainName, out processorHost))
            {
                Log.Trace(() => string.Format("Creating processor host {0}.", appDomainName));

                if (!_options.HasFlag(Options.DoNotCastAppDomains))
                {
                    AppDomain currentDomain = AppDomain.CurrentDomain;
                    AppDomain domain =
                        AppDomain.CreateDomain
                        (
                            appDomainName,
                            null,
                            currentDomain.BaseDirectory,
                            currentDomain.RelativeSearchPath,
                            false
                        );

                    processorHost =
                        (IProcessorHost)domain.CreateInstanceAndUnwrap
                        (
                            Assembly.GetExecutingAssembly().FullName,
                            typeof(ProcessorHost).FullName ?? typeof(ProcessorHost).Name
                        );
                    _sponsor.Register((MarshalByRefObject)processorHost);
                }
                else
                {
                    processorHost = new ProcessorHost();
                }

                _processorHosts.Add(appDomainName, processorHost);

                processorHost.Start(this);
            }

            return processorHost;
        }

        private void StopProcessorHost(string appDomainName)
        {
            Log.Trace(() => string.Format("Stopping processor host {0}.", appDomainName));

            IProcessorHost processorHost;

            if (_processorHosts.TryGetValue(appDomainName, out processorHost))
            {
                processorHost.Stop();

                if (!_options.HasFlag(Options.DoNotCastAppDomains))
                {
                    _sponsor.Unregister((MarshalByRefObject)processorHost);
                    AppDomain.Unload(processorHost.AppDomain);
                }

                _processorHosts.Remove(appDomainName);
                _processorHostStatistics.Remove(appDomainName);
            }
            else
            {
                Log.Warning(string.Format("An attempt has been made to stop the application domain {0} that doesn't exist.", appDomainName));
            }
        }

        private void StopAllProcessors()
        {
            Log.Trace("Stopping all processor hosts.");

            foreach (var appDomainName in _processorHosts.Keys.ToList())
            {
                StopProcessorHost(appDomainName);
            }
        }

        private void RegisterProcess(Process process, bool waiting)
        {
            Log.Trace(() => string.Format("Registering process {0} as {1}.", process.Id, (waiting) ? "waiting" : "running"));

            AssemblyName definingAssemblyName =
                DirectoryUtil.AssemblyNameFromTypeName(process.Workflow);
            string appDomainName = AppDomainNameByAssemblyName(definingAssemblyName);

            if (!_pidToAppDomainMap.ContainsKey(process.Id))
            {
                _pidToAppDomainMap.Add(process.Id, appDomainName);
            }

            if (!waiting)
            {
                GetOrCreateProcessorHost(appDomainName)
                    .LoadAndExecuteProcess(process.Id);
            }
        }

        private void CheckSuspendedProcesses()
        {
            Log.Trace("Checking suspended processes.");

            System.Collections.Generic.IList<string> suspendedProcesses;
            int offset = 0;

            do
            {
                var result =
                    TransactionHelper.Scope
                    (
                        IsolationLevel.ReadCommitted,
                        TimeSpan.FromSeconds(30),
                        () =>
                            _flowerClient
                                .CreateScript()
                                .SetText("@GetSuspended")
                                .AddParameter("processorName", _settings.ProcessorName)
                                .AddParameter("offset", offset)
                                .AddParameter("limit", offset + PROCESSES_CHECK_BATCH_SIZE)
                                .Execute()
                    );

                suspendedProcesses = result.Get<System.Collections.Generic.IList<string>>("pids");
                var timeouts =
                    result
                        .Get<System.Collections.Generic.IList<double>>("timeouts");
                var linkIds = result.Get<System.Collections.Generic.IList<string>>("linkIds");

                if (suspendedProcesses != null)
                {
                    for (int i = 0; i < suspendedProcesses.Count; i++)
                    {
                        //The access is not synchronized because it is assumed
                        //that no any concurent thread is started yet.
                        _suspentions.Add
                        (
                            new Suspention
                            {
                                LinkId = linkIds[i],
                                Pid = suspendedProcesses[i],
                                Until = ScriptHelper.FromJsDate(timeouts[i])
                            }
                        );
                    }
                }

                offset += PROCESSES_CHECK_BATCH_SIZE;
            }
            while (suspendedProcesses != null && suspendedProcesses.Count > 0);
        }

        private void ResumeRunningProcesses()
        {
            Log.Trace("Resuming running processes.");

            long total = 
                TransactionHelper.Scope
                (
                    IsolationLevel.ReadCommitted, 
                    TimeSpan.FromSeconds(30), 
                    () => _flowerClient.Directory.CountChildren(_runningFolderPath, DirectoryEntryTypes.Process)
                );
            long offset = 0;
            long limit = PROCESSES_CHECK_BATCH_SIZE;

            var runningProcesses = new List<DirectoryEntry>();

            do
            {
                TransactionHelper.Scope
                (
                    IsolationLevel.ReadCommitted,
                    TimeSpan.FromSeconds(30),
                    () =>
                    {
                        runningProcesses.AddRange
                        (
                            _flowerClient.Directory.GetChildren
                            (
                                _runningFolderPath,
                                DirectoryEntryTypes.Process,
                                SortProperty.Default,
                                SortOrder.Asc,
                                offset,
                                limit
                            )
                        );
                    }
                );

                offset = limit;
                limit += PROCESSES_CHECK_BATCH_SIZE;
            }
            while (runningProcesses.Count < total);

            foreach (Process runningProcess in runningProcesses)
            {
                try
                {
                    RegisterProcess(runningProcess, false);
                }
                catch (Exception ex)
                {
                    Log.Error(string.Format("Failed to load the process {0}.", runningProcess.Id), ex);
                }
            }
        }

        private void CheckWaitingProcesses()
        {
            Log.Trace("Checking waiting processes.");

            System.Collections.Generic.IList<DirectoryEntry> waitingProcesses;
            int offset = 0;

            do
            {
                waitingProcesses =
                    TransactionHelper.Scope
                    (
                        IsolationLevel.ReadCommitted,
                        TimeSpan.FromSeconds(30),
                        () =>
                            _flowerClient.Directory.GetChildren
                            (
                                _waitingFolderPath,
                                DirectoryEntryTypes.Process,
                                SortProperty.Default,
                                SortOrder.Asc,
                                offset,
                                offset + PROCESSES_CHECK_BATCH_SIZE
                            )
                    );

                foreach (Process waitingProcess in waitingProcesses)
                {
                    RegisterProcess(waitingProcess, true);
                }

                offset += PROCESSES_CHECK_BATCH_SIZE;
            }
            while (waitingProcesses.Count > 0);
        }

        #endregion

        #region IProcessingManager implementation

        public void MarkFinished(string pid)
        {
            MarkFinishedCommand cmd = _markFinishedCommandPool.Get();
            cmd.Pid = pid;
            _actor.Send(cmd);
        }

        public void MarkBroken(string pid)
        {
            MarkBrokenCommand cmd = _markBrokenCommandPool.Get();
            cmd.Pid = pid;
            _actor.Send(cmd);
        }

        public void MarkWaiting(string pid)
        {
            MarkWaitingCommand cmd = _markWaitingCommandPool.Get();
            cmd.Pid = pid;
            _actor.Send(cmd);
        }

        public void MarkRunning(string pid)
        {
            MarkRunningCommand cmd = _markRunningCommandPool.Get();
            cmd.Pid = pid;
            _actor.Send(cmd);
        }

        public void MarkSuspended(string pid, DateTime until, string linkId)
        {
            _suspentionsLock.EnterWriteLock();
            try
            {
                _suspentions.Add
                (
                    new Suspention 
                    {
                        LinkId = linkId,
                        Pid = pid,
                        Until = until
                    }
                );
            }
            finally
            {
                _suspentionsLock.ExitWriteLock();
            }
        }

        #endregion

        #region IProcessor implementation

        private static WindowsIdentity GetCurrentUser()
        {
            return
                (ServiceSecurityContext.Current != null) ?
                    ServiceSecurityContext.Current.WindowsIdentity :
                    WindowsIdentity.GetAnonymous();
        }

        public void CheckPendingProcesses()
        {
            if (_isSuspended) return;

            CheckPendingProcesses(true);
        }

        private void CheckPendingProcesses(bool resumeActors)
        {
            if (_isSuspended) return;

            try
            {
                if (!EnsureInitialized())
                {
                    return;
                }

                if (resumeActors)
                {
                    _actor.Resume();
                }

                if (!_actor.IsSuspended)
                {
                    CheckPendingProcessesCommand cmd = _checkPendingProcessesCommandPool.Get();
                    cmd.ResumeActors = resumeActors;
                    _actor.Send(cmd);
                }
            }
            catch (Exception ex)
            {
                Log.Error("CheckPendingProcesses failed.", ex);
            }
        }

        private void CheckRunningStateChangeAllowed()
        {
            WindowsIdentity user = GetCurrentUser();
            var principal = new WindowsPrincipal(user);

            bool allow =
                _settings.AllowStartStopToUsersList
                    .Any(u => user.Name.Equals(u, StringComparison.OrdinalIgnoreCase)) ||
                _settings.AllowStartStopToRolesList
                    .Any(principal.IsInRole);

            if (!allow)
            {
                Log.Error(string.Format("Running state change denied for user {0}.", user.Name));
                throw new SecurityException("Access denied");
            }
        }

        public void IsRunning(bool state)
        {
            try
            {
                CheckRunningStateChangeAllowed();
            }
            catch (Exception ex)
            {
                if (!(ex is SecurityException)) 
                {
                    Log.Error("Failed to check permissions.", ex);
                }

                throw;
            }

            try
            {
                if (_isSuspended)
                {
                    if (!state) return;

                    ResumeRunningProcesses();
                    CheckWaitingProcesses();
                    CheckSuspendedProcesses();

                    _isSuspended = false;

                    CheckPendingProcesses();

                    Log.Info("Processor is resumed.");
                }
                else
                {
                    if (state) return;

                    _isSuspended = true;

                    _actor.WaitAllMessages(MaxActorWaitTime);

                    StopAllProcessors();
                    _pidToAppDomainMap.Clear();

                    Log.Info("Processor is suspended.");
                }
            }
            catch(Exception ex)
            {
                Log.Error(string.Format("Failed to {0} processor.", (state) ? "resume" : "suspend"), ex);
            }
        }

        #endregion
    }
}
